package com.yunhe.transfer.mqtt;

import cn.beepower.domain.bus.MqttDataBus;
import cn.beepower.domain.bus.ThreadMode;
import cn.beepower.domain.message.BeeMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.model.authority.Station;
import com.yunhe.transfer.client.AuthorityClient;
import com.yunhe.common.constant.MqttConsts;
import com.yunhe.common.model.authority.StationDTO;
import com.yunhe.transfer.service.data.HeartbeatService;
import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
import com.yunhe.common.util.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 心跳MQTT处理类
 * @author liuronglei
 */
@Component
public class HeartbeatMqttHandler {
    private static final ObjectMapper mapper = new ObjectMapper();
    private Logger logger = LoggerFactory.getLogger(HeartbeatMqttHandler.class);
    @Autowired
    @Qualifier("MqttDataBus")
    private MqttDataBus dataBus;
    @Autowired
    private HeartbeatService heartbeatService;
    @Autowired
    private AuthorityClient authorityClient;
    @Autowired
    @Qualifier("allStations")
    private List<Station> stations;

    @PostConstruct
    public void init() {
        if (CollectionUtil.isNotEmpty(stations)) {
            for (Station station : stations) {
                Long stationCode = station.getId();
                if (stationCode == null) {
                    continue;
                }
                subscribe(stationCode);
            }
        }
    }

    /**
     * 订阅消息
     * @param stationCode 电站代码
     */
    public void subscribe(Long stationCode) {
        dataBus.subscribe(MqttConsts.TOPIC.HeartbeatClient.value(stationCode), 0, ThreadMode.Async, msg -> {
            BeeMessage message = msg.getM();
            String topic = message.getTopic();
            logger.info("接收心跳：{}", topic);
            heartbeatService.updateHeartbeatTimeMillis(MqttConsts.TOPIC.HeartbeatClient.getStationCode(topic));
            return null;
        });
    }

    // 每5秒执行心跳上传
    // 或者取yml配置:custom.collect.heartbeat.frequency
    @SchedulerLock(name = "HeartbeatClientMqttTask", lockAtMostFor = "PT3S")
    @Scheduled(cron = "0/5 * * * * ? ")
    public void heartbeattClientMqttTask() throws Exception {
        Map<String, Long> sendResult = new HashMap<>();
        sendResult.put("heartbeat", System.currentTimeMillis());
        String result = mapper.writeValueAsString(sendResult);
        List<Station> stations = authorityClient.getStations(null).getResults();
        // 循环每个需要心跳判断的电站
        if (CollectionUtil.isNotEmpty(stations)) {
            for (Station station : stations) {
                Long stationCode = station.getId();
                // 如果电站code为空，或者不使用心跳逻辑，则跳过
                if (stationCode == null || station.getUseHeartbeat() == null || !station.getUseHeartbeat()) {
                    continue;
                }
                // 主题格式：heartbeat/server/1104
                dataBus.publish(MqttConsts.TOPIC.HeartbeatServer.value(station.getId()), result.getBytes(), 0);
            }
        }
    }
}
